package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.orderPeoCount;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/2 8:52
 * @From: GoodGoodStudyDayDayUp
 *
 *  Kafka-Connector 会过滤掉主题中的 null 数据，因此订单明细主题不存在为 null 的数据，直接转换数据结构即可
 *
 *          doris 建表语句：
 *                        drop table if exists dws_user_order_unique_window;
 *                        create table if not exists dws_user_order_unique_window
 *                        (
 *                            `stt`                            DATETIME comment '窗口起始时间',
 *                            `edt`                            DATETIME comment '窗口结束时间',
 *                            `cur_date`                       DATE comment '当天日期',
 *                            `order_unique_people_ct`         BIGINT replace comment '独立下单用户数'
 *                        ) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
 *                        comment "用户主题下单汇总表"
 *                        partition by range(`cur_date`)()
 *                        distributed by hash(`stt`) buckets 10 properties (
 *                          "replication_num" = "3",
 *                          "dynamic_partition.enable" = "true",
 *                          "dynamic_partition.time_unit" = "DAY",
 *                          "dynamic_partition.start" = "-1",
 *                          "dynamic_partition.end" = "1",
 *                          "dynamic_partition.prefix" = "par",
 *                          "dynamic_partition.buckets" = "10",
 *                          "dynamic_partition.hot_partition_num" = "1"
 *                        );
 */
public class DwsUserOrderCountWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {

        new DwsUserOrderCountWindow_lixuanfeng().init(3104,2,
                "DwsUserOrderCountWindow_lixuanfeng",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        SingleOutputStreamOperator<orderPeoCount> reduce = stream
                .map(jsonStr -> JSON.parseObject(jsonStr))
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String userId = value.getString("user_id");

                        return userId != null;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner((json, ts) -> json.getLong("ts") * 1000))
                .keyBy(jsonStr -> jsonStr.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, orderPeoCount>() {

                    private ValueState<String> lastOrderDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastOrderDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastOrderDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<orderPeoCount> out) throws Exception {

                        Long orderPeopleCt = 0L;

                        String lastDate = lastOrderDate.value();
                        Long ts = value.getLong("ts");
                        String nowOrderDate = AtguiguUtil.toDate(ts);

                        if (lastDate == null || (lastDate != null && !nowOrderDate.equals(lastDate))) {
                            orderPeopleCt = 1L;
                            lastOrderDate.update(nowOrderDate);
                        }

                        out.collect(orderPeoCount.builder()
                                .orderUniquePeopleCt(orderPeopleCt)
                                .build());
                    }
                })
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<orderPeoCount>() {
                            @Override
                            public orderPeoCount reduce(orderPeoCount value1, orderPeoCount value2) throws Exception {
                                value1.setOrderUniquePeopleCt(value1.getOrderUniquePeopleCt() + value2.getOrderUniquePeopleCt());
                                return value1;
                            }
                        },
                        new ProcessAllWindowFunction<orderPeoCount, orderPeoCount, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<orderPeoCount> elements, Collector<orderPeoCount> out) throws Exception {
                                orderPeoCount data = elements.iterator().next();

                                String start = AtguiguUtil.toDate(context.window().getStart());
                                String end = AtguiguUtil.toDate(context.window().getEnd());

                                String curdate = AtguiguUtil.toDate(System.currentTimeMillis());

                                data.setStt(start);
                                data.setStt(end);
                                data.setCurDate(curdate);

                                out.collect(data);
                            }
                        }
                );
       // reduce.print("okokokokokokokokokokok");



        //TODO 写入 doris

        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_user_order_unique_window";

        reduce.map( pojo -> {
            SerializeConfig config = new SerializeConfig();
            config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
            return JSON.toJSONString(pojo, config); })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
    }

}
